package mq.rabbit.demo;

import java.io.IOException;

import com.rabbitmq.client.*;
import mq.rabbit.common.Base;
import org.junit.Test;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 第一种模型（直连）
 *
 * P：生产者，也就是要发送消息的程序
 * C：消费者，会一直等待消息的到来
 * queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。
 */
public class OneToOne extends Base {

    // 生产消息
    @Test
    public void testProviderSendMessage() throws IOException, TimeoutException {

        // 获取连接对象
        Connection connection = connectionFactory.newConnection();

        // 获取连接中通道对象
        Channel channel = connection.createChannel();

        // 通道绑定对应消息队列
        // 参数1：队列名称，如果队列不存在会自动创建
        // 参数2：用来定义队列特性是否要持久化
        // 参数3：是否独占队列，表示只有当前连接可用该队列
        // 参数4：是否在消费完成后自动删除队列
        // 参数5：额外附加参数
        channel.queueDeclare("hello", false, false, false, null);

        // 发布消息
        // 参数1：交换器名称
        // 参数2：队列名称
        // 参数3：传递消息额外设置
        // 参数4：消息的具体内容
        channel.basicPublish("", "hello", null, "hello rabbitmq".getBytes());

        // 关闭
        channel.close();
        connection.close();
    }

    @Test
    public void testConsumerReceiveMessage() throws IOException, TimeoutException, InterruptedException {
        // 通道绑定对应消息队列
        // 参数1：队列名称，如果队列不存在会自动创建
        // 参数2：用来定义队列特性是否要持久化
        // 参数3：是否独占队列，表示只有当前连接可用该队列
        // 参数4：是否在消费完成后自动删除队列
        // 参数5：额外附加参数
        Connection connection = connectionFactory.newConnection();

        // 创建通道
        Channel channel = connection.createChannel();

        // 通道绑定对应消息队列（注意参数要与生产者对应）
        // 参数1：队列名称，如果队列不存在会自动创建
        // 参数2：用来定义队列特性是否要持久化
        // 参数3：是否独占队列，表示只有当前连接可用该队列
        // 参数4：是否在消费完成后自动删除队列
        // 参数5：额外附加参数
        channel.queueDeclare("hello", false, false, false, null);

        // 消费消息
        // 参数1：消息哪个队列的消息
        // 参数2：开始消息的自动确认机制
        // 参数3：消费时的回调接口
        channel.basicConsume("hello", true, new DefaultConsumer(channel) {
            // 参数：body 消息队里中取出的消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                System.out.println("输出队列中的内容>>>>>>>>>: " + new String(body));
            }
        });

        System.out.println(" ============== ");
        TimeUnit.MINUTES.sleep(1);
        // 如果没关闭，消费者会一直等待
        // 关闭，不会等结果出来（不推荐）
        //channel.close();
        //connection.close();
    }
}